package com.example.springbootusecase.es;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.example.springbootusecase.config.ElasticsearchConfig;
import com.google.protobuf.InvalidProtocolBufferException;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Service
public class CanalToElasticsearch {

    @Resource
    private RestHighLevelClient esClient;

    // ✅ 让 Canal 在 Spring Boot 启动后自动运行
    @PostConstruct
    public void startCanalListener() {
        new Thread(this::listenToMySQL).start();
    }
    public void listenToMySQL() {
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1", 11111), "example", "", "");
        connector.connect();
        connector.subscribe("db2021.products"); // 监听 test 库的 products 表

        while (true) {
            //从 Canal 服务器获取最多 100 条变更数据
            Message message = connector.get(100);
            List<CanalEntry.Entry> entries = message.getEntries();

            for (CanalEntry.Entry entry : entries) {
                //只处理 ROWDATA 类型（行级变更），跳过事务日志、DDL 语句等
                if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                    CanalEntry.RowChange rowChange = null;
                    try {
                        rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                    } catch (InvalidProtocolBufferException e) {
                        e.printStackTrace();
                    }

                    for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                        switch (rowChange.getEventType()) {
                            case INSERT:
                                syncToElasticsearch(rowData.getAfterColumnsList());
                                break;
                            case UPDATE:
                                updateElasticsearch(rowData.getAfterColumnsList());
                                break;
                            default:
                                break;
                        }
                    }
                }
            }
        }
    }

    // ✅ 处理新增（INSERT）数据同步
    private void syncToElasticsearch(List<CanalEntry.Column> columns) {
        Map<String, Object> doc = new HashMap<>();
        String id = "";

        for (CanalEntry.Column column : columns) {
            doc.put(column.getName(), column.getValue());
            if (column.getName().equals("id")) {
                id = column.getValue();
            }
        }

        try {
            IndexRequest indexRequest = new IndexRequest("products").id(id).source(doc);
            esClient.index(indexRequest, RequestOptions.DEFAULT);
            System.out.println("✅ 成功同步数据到 Elasticsearch：" + doc);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    // ✅ 处理修改（UPDATE）数据同步
    private void updateElasticsearch(List<CanalEntry.Column> columns) {
        Map<String, Object> doc = new HashMap<>();
        String id = "";

        for (CanalEntry.Column column : columns) {
            doc.put(column.getName(), column.getValue());
            if (column.getName().equals("id")) {
                id = column.getValue();
            }
        }

        try {
            IndexRequest updateRequest = new IndexRequest("products").id(id).source(doc);
            esClient.index(updateRequest, RequestOptions.DEFAULT);
            System.out.println("✅ 修改数据同步到 ES：" + doc);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


}
